package com.streaming.demo

import com.alibaba.fastjson.JSON
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext.rddToFileName
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext, Time}
import org.json4s.ShortTypeHints
import org.json4s.jackson.Serialization

/**
  * @author td
  * @date 2018/4/2
  */
object SparkStreamingHdfsDemo {

  case class Device(brand:String)

  def test(sparkContext: SparkContext): Unit ={
    implicit val formats = Serialization.formats(ShortTypeHints(List()));
    val input = sparkContext.textFile("hdfs://master:8020/user/fingerprintAndroid/20180309/");

    /*input.collect().foreach(x=>{
      var c =
    })*/
    input.collect().foreach(x=>{
       val json = JSON.parseObject(x);
       println(json.get("brand"));
    })
  }

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("demo");
    val sparkContext = new SparkContext(conf);
    val scc = new StreamingContext(sparkContext,Seconds(1));

    val input = scc.textFileStream("hdfs://master:8020/user/fingerprintAndroid/20180309/");

    val listRdd = input.map(jsonString=>{
      val obj = JSON.parseObject(jsonString);
      val brand = obj.get("brand");
      (brand,1);
    });

    val reduceByKey = listRdd.reduceByKey(_+_);
    //reduceByKey.

    reduceByKey.saveAsTextFiles("hdfs://master:8020/user/streaming/result");


   /* val saveFunc = (rdd: RDD[T], time: Time) => {
      val file = rddToFileName(prefix, suffix, time)
      rdd.saveAsTextFile(file)
    }*/



   // reduceByKey.saveAsTextFiles("hdfs://master:8020/user/streaming/result",".result");

    scc.start();
    scc.awaitTermination();




   // test(sparkContext);

    /*implicit val formats = Serialization.formats(ShortTypeHints(List()));


    // 设置配置
    val conf = new SparkConf().setMaster("local[2]").setAppName("sparkDemo");


    // 设置stremingContext 上下文
    val ssc = new StreamingContext(conf,Seconds(1));

    val lines = ssc.textFileStream("hdfs://master:8020/user/fingerprintAndroid/20180321/");*/

   /* lines.flatMap(jsonSting=>{

    });*/








   /* lines.flatMap(jsonString=> {
      val obj =

    });*/

    /*lines.flatMap(value=>{
      println(value._1+"--"+value._2);
    });*/


   // Serialization.formats(ShortTypeHints)



  //  val wordsRdd = lines.flatMap(_.split(" ")).map((_,1));

 //   val reduceRdd = wordsRdd.reduceByKey(_+_);

 //   reduceRdd.print();

    //ssc.start();

  //  ssc.awaitTermination();
  }

}
